package com.tcp2.net;

import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;

import com.tcp2.exception.LinkException;
import com.tcp2.server.DxsConfig;
import com.tcp2.server.MessageDispatcher;
import com.tcp2.server.ProtoMsg;
import com.tcp2.ui.MainFrame;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 从链路
 * Author: 杨俊辉
 * Time: 2014-09-23 09:53
 * Copyright (C) 2014 Xiamen Yaxon Networks CO.,LTD.
 */
public class SlaveLink {
    @Autowired
    private MasterLink masterLink;
    protected static Logger logger = LoggerFactory.getLogger(SlaveLink.class);
    private static final int CONNECT_TIMEOUT_MILLIS = 20000;
    private DxsConfig dxsConfig;
    private MessageDispatcher messageDispatcher;
    private SlaveLinkHandler slaveLinkHandler;
    private Bootstrap bootstrap;
    private EventLoopGroup ioThreadPool;
    private int gnssCenterId; //下级平台接入码
    private Channel channel;

    private String downLinkIp ;
    private  int downLinkPort;
    private int verifyCode;
    private volatile int linkState = 0; //链路状态
    private static final int LS_STARTED = 1;//已启动
    private static final int LS_STOPPED = 2;//已停止

    private volatile int connectState = CS_CLOSED; //连接状态
    private static final int CS_CLOSED = 0;
    private static final int CS_CONNECTING = 1;
    private static final int CS_CONNECTED = 2;
    private static final int CS_AUTHENTICATED = 3;
    private static final int CS_CLOSING = 4;

    private AtomicLong lastContactTime = new AtomicLong(0);
    private final Lock connectLock = new ReentrantLock(false);

    private ProtoLogger protoLogger = new ProtoLogger("从链路");


    private Thread loopThread;

    class LoopThread extends Thread {
        LoopThread() {
            setName("slave-link-loop-thread");
            setDaemon(true);
        }

        @Override
        public void run() {
            int ticks = 0;
            long t = 0;
            while (true) {
                ++ticks;
                t = System.currentTimeMillis();
                try {
                    if (ticks % 30 == 0) { //发送心跳
                        if (connectState == CS_AUTHENTICATED) {
                            slaveLinkHandler.sendLinkTestReq(channel);
                        }

                    }

                    if (ticks % 5 == 0) { //连接状态检查
                        if (linkState != LS_STOPPED) {
                            if (connectState == CS_CLOSED) {//发起连接请求
                               // openConnect(downLinkIp, downLinkPort);//屏蔽此避免出现重启下级后出现从链路连接后又断开，断开后又连接的问题
                            } else if (connectState == CS_CONNECTING) {
                                if (t - lastContactTime.get() > CONNECT_TIMEOUT_MILLIS) {
                                    logger.warn("连接超时，即将关闭连接,下级平台接入码：{}",gnssCenterId);
                                    closeConnect();

                                }
                            } else if (connectState == CS_CONNECTED) {
                                if (t - lastContactTime.get() > 30000) {
                                    logger.warn("连接后登录鉴权超时，即将关闭连接,下级平台接入码：{}",gnssCenterId);
                                    closeConnect();

                                }
                            } else if (connectState == CS_AUTHENTICATED) {
                                if (t - lastContactTime.get() > 180000) {
                                    logger.warn("链路保持超时，即将关闭连接,下级平台接入码：{}",gnssCenterId);
                                    closeConnect();

                                }
                            }
                        }
//                        else {
//                            if (masterLink != null){
//                                masterLink.stop();
//                                masterLink = null;
//                            }
//
//                        }

                        if(connectState==CS_CONNECTING||connectState==CS_AUTHENTICATED){
                            MainFrame.SlaveStatus=true;
                        }
                        else if(connectState==CS_CLOSED){
                            MainFrame.SlaveStatus=false;
                        }


                    }

                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    break;
                } catch (Exception e) {
                    logger.warn("从链路后台线程异常,下级平台接入码：{}",gnssCenterId, e);
                }
            }
            logger.info("线程({})已停止,下级平台接入码：{}", getName(),gnssCenterId);
        }
    }

    public void setDxsConfig(DxsConfig dxsConfig) {
        this.dxsConfig = dxsConfig;
    }

    public MessageDispatcher getMessageDispatcher() {
        return messageDispatcher;
    }

    public void setMessageDispatcher(MessageDispatcher messageDispatcher) {
        this.messageDispatcher = messageDispatcher;
        messageDispatcher.setSlaveLink(this);
    }

    public static Logger getLogger() {
        return logger;
    }

    public static void setLogger(Logger logger) {
        SlaveLink.logger = logger;
    }

    public String getDownLinkIp() {
        return downLinkIp;
    }

    public void setDownLinkIp(String downLinkIp) {
        this.downLinkIp = downLinkIp;
    }

    public int getDownLinkPort() {
        return downLinkPort;
    }

    public void setDownLinkPort(int downLinkPort) {
        this.downLinkPort = downLinkPort;
    }

    public int getVerifyCode() {
        return verifyCode;
    }

    public void setVerifyCode(int verifyCode) {
        this.verifyCode = verifyCode;
    }

    public int getGnssCenterId() {
        return gnssCenterId;
    }

    public void setGnssCenterId(int gnssCenterId) {
        this.gnssCenterId = gnssCenterId;
    }

    /**
     * 开启通道
     *
     * @throws Exception
     */
    public void start() throws Exception {
        if (linkState == LS_STARTED) {
            return;
        }


        try {
            ioThreadPool = new NioEventLoopGroup();
            bootstrap = new Bootstrap();
            if (slaveLinkHandler == null) {
                slaveLinkHandler = new SlaveLinkHandler(this, messageDispatcher, dxsConfig);
            }

            bootstrap.group(ioThreadPool)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(protoLogger);
                            p.addLast(new ProtoMsgDecoder(dxsConfig));
                            p.addLast(new ProtoMsgEncoder(dxsConfig));
                            p.addLast(slaveLinkHandler);
                        }
                    });
            bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, CONNECT_TIMEOUT_MILLIS);
            bootstrap.option(ChannelOption.TCP_NODELAY, true);
            bootstrap.option(ChannelOption.SO_REUSEADDR, true);

            openConnect(downLinkIp, downLinkPort);

            loopThread = new LoopThread();
            loopThread.start();

            linkState = LS_STARTED;

            logger.info("从链路服务已启动,下级平台接入码：{}", gnssCenterId);
        } catch (Exception e) {
            logger.warn("从链路服务已启动,下级平台接入码：{}",gnssCenterId, e);
        }
    }

    /**
     * 关闭从链路服务
     */
    public void stop() {
        if (linkState == LS_STOPPED) {
            return;
        }

        try {
            closeConnect();

            if (loopThread != null) {
                loopThread.interrupt();
                loopThread = null;
            }

            if (ioThreadPool != null) {
                ioThreadPool.shutdownGracefully();
                ioThreadPool = null;
            }

            if (bootstrap != null) {
                bootstrap = null;
            }

            linkState = LS_STOPPED;
            logger.info("从链路服务已关闭,下级平台接入码：{}",gnssCenterId);
        } catch (Exception e) {
            logger.warn("关闭从链路服务失败,下级平台接入码：{}",gnssCenterId, e);
        }
    }

    protected void openConnect(String downHost, int downPort) {
        logger.info("正在发起从链路连接({}:{})..", downHost, downPort);
        connectLock.lock();
        try {
            if (connectState != CS_CLOSED) {
                return;
            }

            bootstrap = new Bootstrap();
            if (slaveLinkHandler == null) {
                slaveLinkHandler = new SlaveLinkHandler(this, messageDispatcher, dxsConfig);
            }

            bootstrap.group(ioThreadPool)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(protoLogger);
                            p.addLast(new ProtoMsgDecoder(dxsConfig));
                            p.addLast(new ProtoMsgEncoder(dxsConfig));
                            p.addLast(slaveLinkHandler);
                        }
                    });
            bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, CONNECT_TIMEOUT_MILLIS);
            bootstrap.option(ChannelOption.TCP_NODELAY, true);
            bootstrap.option(ChannelOption.SO_REUSEADDR, true);

            ChannelFuture cf = bootstrap.connect(downHost, downPort);
            connectState = CS_CONNECTING;
            lastContactTime.set(System.currentTimeMillis());
            cf.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (!future.isSuccess()) {
                        logger.warn("开启从链路连接异常,下级平台接入码：{}",gnssCenterId ,future.cause());
                        connectState = CS_CLOSED;
                    }
                }
            });
        } finally {
            connectLock.unlock();
        }
    }

    protected void closeConnect() {
        connectLock.lock();
        try {
            if (connectState == CS_CLOSED) {
                return;
            }

            if (channel != null) {
                channel.close();
                connectState = CS_CLOSING;
                lastContactTime.set(System.currentTimeMillis());
            }
        } finally {
            connectLock.unlock();
        }
    }

    protected void connnected(Channel ch) {
        connectLock.lock();
        connectState = CS_CONNECTED;
        this.channel = ch;
        lastContactTime.set(System.currentTimeMillis());
        connectLock.unlock();
    }

    protected void authenticated() {
        connectLock.lock();
        connectState = CS_AUTHENTICATED;
        lastContactTime.set(System.currentTimeMillis());
        connectLock.unlock();
    }

    protected void disconnected() {
        connectLock.lock();
        connectState = CS_CLOSED;
        channel = null;
        lastContactTime.set(0);
        connectLock.unlock();
    }

    protected void checkAlive() {
        lastContactTime.set(System.currentTimeMillis());
    }


    public void sendMessage(ProtoMsg msg) throws LinkException {
        try {
            if (msg == null) {
                throw new RuntimeException("待发送的消息不能为null,下级平台接入码："+gnssCenterId);
            }

            if (!isActive()) {
                throw new RuntimeException("从链路未建立,下级平台接入码："+gnssCenterId);
            }

            slaveLinkHandler.sendMessage(channel, msg).sync();
        } catch (Exception e) {
            throw new LinkException("通过从链路发送消息失败,下级平台接入码："+gnssCenterId, e);
        }
    }

    public ListenableFuture<Void> sendMessage0(ProtoMsg msg) {
        final SettableFuture<Void> sf = SettableFuture.create();

        try {
            if (msg == null) {
                throw new RuntimeException("待发送的消息不能为null,,下级平台接入码："+gnssCenterId);
            }

            if (connectState != CS_AUTHENTICATED) {
                throw new RuntimeException("从链路未建立,下级平台接入码："+gnssCenterId);
            }

            slaveLinkHandler.sendMessage(channel, msg).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture cf) throws Exception {
                    if (cf.isSuccess()) {
                        sf.set(null);
                    } else {
                        logger.warn("通过从链路发送消息发生异常,下级平台接入码：{}",gnssCenterId, cf.cause());
                        sf.setException(new RuntimeException("通过从链路发送消息发生异常,下级平台接入码："+gnssCenterId, cf.cause()));
                    }
                }
            });
        } catch (Exception e) {
            sf.setException(e);
        }

        return sf;
    }

    public boolean isStopped() {
        return linkState == LS_STOPPED;
    }

    public boolean isActive() {
        if (linkState == LS_STARTED && connectState == CS_AUTHENTICATED) {
            return true;
        } else {
            return false;
        }
    }
}
